package cn.ipanel.bigdata.dw

import cn.ipanel.bigdata.boot.config.Configuration
import cn.ipanel.bigdata.boot.date.{Day, Hour, Month, Quarter, Week, Year}
import cn.ipanel.bigdata.boot.logger.Logger
import cn.ipanel.bigdata.comm.Parameter.ODS_DB_NAMESPACE
import cn.ipanel.bigdata.dw
import cn.ipanel.bigdata.utils.{Dictionary, Util}
import org.apache.spark.sql.{DataFrame, Row}
import org.apache.spark.sql.functions.{col, lit}

/**
 * Author: lzz
 * Date: 2022/3/24 19:14
 */
package object ods {

  final lazy val BUILD: String = "create database if not exists " + ODS_DB_NAME
  final lazy val ODS_DB_NAME: String = Configuration.getParam(ODS_DB_NAMESPACE)

  final lazy val T_REPORT   = Report
  final lazy val T_REPORT_BY_HOUR   = ReportByHour
  final lazy val T_HDFS_LOG = HdfsLog


  final lazy val Tables: Seq[AbsOdsHive] = Seq(T_REPORT)

  trait OdsBasicColumnDefine {

    final val F_DATE_TIME    : String = "f_date_time"         // yyyy-MM-dd HH:mm:ss
    final val F_DEVICE_ID    : String = "f_device_id"         // 设备id，唯一
    final val F_REGION       : String = "f_region"            // （市）区域ID
    final val F_EXTRA        : String = "f_extra"             // 扩展参数，后续如果增加字段，可以放这里面
    final val F_DATE         : String = Dictionary.F_DATE     // 分区字段
  }

  abstract class AbsOdsHive(tbName: String, dbName: String = ODS_DB_NAME)
          extends dw.AbsDwHive(dbName, tbName)
            with OdsBasicColumnDefine {

    def find(day: Day, range: Int = 0): DataFrame = {
      if (range == 0) {
        load.filter(col(F_DATE) === lit(day.toDate))
      } else {
        val fDay = if (range < 0) day.prev(Math.abs(range)).toDate else day.toDate
        val lDay = if (range > 0) day.next(Math.abs(range)).toDate else day.toDate
        load.filter(col(F_DATE) >= lit(fDay) and col(F_DATE) <= lit(lDay))
      }
    }

    def findByHour(hour: Hour, range: Int = 0): DataFrame = {
      if (range == 0) {
        load.filter(col(F_DATE) === lit(hour.toFullHour))
      } else {
        val sHour = if (range < 0) hour.prev(Math.abs(range)).toDate else hour.toFullHour
        val eHour = if (range > 0) hour.next(Math.abs(range)).toDate else hour.toFullHour
        load.filter(col(F_DATE) >= lit(sHour) and col(F_DATE) <= lit(eHour))
      }
    }

    def drop(day: Day, range: Int = 0): Unit = {
      if (range == 0) {
        exec(s"ALTER TABLE $getDBName.$getTBName DROP IF EXISTS PARTITION($F_DATE=${day.toDate})")
      } else {
        val fDay = if (range < 0) day.prev(Math.abs(range)).toDate else day.toDate
        val lDay = if (range > 0) day.next(Math.abs(range)).toDate else day.toDate
        exec(s"ALTER TABLE $getDBName.$getTBName DROP IF EXISTS PARTITION($F_DATE>=$fDay, $F_DATE<=$lDay)")
      }
    }

    def dropByHour(hour: Hour, range: Int = 0): Unit = {
      if (range == 0) {
        exec(s"ALTER TABLE $getDBName.$getTBName DROP IF EXISTS PARTITION($F_DATE=${hour.toFullHour})")
      } else {
        val sHour = if (range < 0) hour.prev(Math.abs(range)).toDate else hour.toFullHour
        val eHour = if (range > 0) hour.next(Math.abs(range)).toDate else hour.toFullHour
        exec(s"ALTER TABLE $getDBName.$getTBName DROP IF EXISTS PARTITION($F_DATE>=$sHour, $F_DATE<=$eHour)")
      }
    }

    def createTable: AbsOdsTable

    override def getTBColumns: Seq[String] = {
      Seq(F_DATE_TIME, F_DEVICE_ID, F_REGION) ++ extraColumnNames ++ Seq(F_EXTRA, F_DATE)
    }

    override def buildTable: String = {
      s"""
         | CREATE TABLE IF NOT EXISTS $getDBName.$getTBName (
         |  $F_DATE_TIME STRING,
         |  $F_DEVICE_ID STRING,
         |  $F_REGION STRING,
         |  $extraColumnBuild,
         |  $F_EXTRA MAP<STRING, STRING>
         | )
         | PARTITIONED by($F_DATE INT)
         | ROW FORMAT DELIMITED FIELDS TERMINATED BY $dataRowFormatter
         | MAP KEYS TERMINATED BY $dataMapFormatter
         | STORED AS $storageFormatter
         |""".stripMargin
//      CLUSTERED BY($F_USER, $F_DEVICE) INTO $splitBucketsNums BUCKETS
    }
  }

  abstract class AbsOdsTable(var f_time: String,
                             var f_device_id: String,
                             var f_region: String,
                             var f_extra: Map[String, String],
                             var f_date: Int)
          extends AbsDwTable
            with OdsBasicColumnDefine
            with Serializable
            with Product {

    import Util._

    lazy val _PARAM: Array[Any] = Array(f_time, f_device_id, f_region,
      f_extra, f_date) ++ toParam

    def _init_(row: Row): AbsDwTable = {
      f_time          = rowTo[String](row, F_DATE_TIME, NONE)
      f_device_id     = rowTo[String](row, F_DEVICE_ID, NONE)
      f_region        = rowTo[String](row, F_REGION, NONE)
      f_extra         = rowTo[Map[String, String]](row, F_EXTRA, EMPTY_MAP)
      f_date          = rowTo[Int](row, F_DATE, INT0)
      this
    }

    def toParam: Array[Any]

    override def productElement(n: Int): Any = _PARAM(n)
    override def productArity: Int = _PARAM.length
    override def canEqual(that: Any): Boolean = _PARAM.contains(that)
  }

  abstract class AbsOdsTable2(var f_time: String,
                              var f_device: String,
                              var f_region: String,
                              var f_extra: Map[String, String],
                              var f_date: Int)
    extends AbsDwTable
      with OdsBasicColumnDefine
      with Serializable
      with Product {

    import Util._

    lazy val _PARAM: Array[Any] = Array(f_time, f_device, f_region) ++ toParam2 ++ toParam ++ Array(f_extra, f_date)

    def _init_(row: Row): AbsDwTable = {
      f_time      = rowTo[String](row, "", NONE)
      f_device    = rowTo[String](row, "", NONE)
      f_region    = rowTo[String](row, F_REGION, NONE)
      f_extra     = rowTo[Map[String, String]](row, F_EXTRA, EMPTY_MAP)
      f_date      = rowTo[Int](row, F_DATE, INT0)
      this
    }

    def toParam: Array[Any]

    def toParam2: Array[Any] = Array()

    override def productElement(n: Int): Any = _PARAM(n)
    override def productArity: Int = _PARAM.length
    override def canEqual(that: Any): Boolean = _PARAM.contains(that)
  }
}